package com.cwh.taskcenter.mq.impl;

import com.alibaba.fastjson.JSON;
import com.cwh.taskcenter.mq.MessageProducer;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;

/**
 * @author cwh
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class KafkaMessageProducer implements MessageProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public boolean send(String topic, String key, Object message) {
        String payload = serializeMessage(message);
        kafkaTemplate.send(topic, key, payload);
        log.info("消息发送成功: topic={}, key={}", topic, key);
        return true;
    }

    @Override
    public void sendAsync(String topic, String key, Object message, SendCallback callback) {
        String payload = serializeMessage(message);
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, key, payload);
        future.whenComplete((result, ex) -> {
            if (ex != null) {
                callback.onFailure(ex); // 发送失败时通知回调
            } else {
                callback.onSuccess(); // 发送成功时通知回调
            }
        });
    }

    private String serializeMessage(Object message) {
        if (message instanceof String) {
            return (String) message;
        }
        return JSON.toJSONString(message);
    }
}
